package com.nightsoul.hadoop1.test.format;

import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import com.nightsoul.hadoop1.test.OldJobBuilder;
import com.nightsoul.hadoop1.test.junit.NcdcRecordParser;

public class PartitionByStationUsingMultipleOutputFormat extends Configured implements Tool {
	/**
	 * StationMapper 将stationId从记录中抽取出来并将它作为键，这样可以将每一个气象站的
	 * 数据切分到同一个分区，默认使用的Partitioner为HashPartitioner
	 * @author zj
	 *
	 */
	static class StationMapper extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
		private NcdcRecordParser parser = new NcdcRecordParser();
		@Override
		public void map(LongWritable key, Text value,
				OutputCollector<Text, Text> output, Reporter reporter)
				throws IOException {
			
			parser.parse(value);
			output.collect(new Text(parser.getStationId()), value);
		}
		
	}
	
	static class StationReducer extends MapReduceBase implements Reducer<Text, Text, NullWritable, Text> {

		@Override
		public void reduce(Text key, Iterator<Text> values,
				OutputCollector<NullWritable, Text> output, Reporter reporter)
				throws IOException {
			while(values.hasNext()) {
				output.collect(NullWritable.get(), values.next());
			}
		}
		
	}
	
	static class StationNameMultipleTextOutputFormat extends MultipleTextOutputFormat<NullWritable, Text> {
		private NcdcRecordParser parser = new NcdcRecordParser();
		
		@Override
		protected String generateFileNameForKeyValue(NullWritable key, Text value, String name) {
			parser.parse(value);
			return parser.getStationId();
		}
		
	}
	
	
	
	@Override
	public int run(String[] args) throws Exception {
		JobConf conf = OldJobBuilder.parseInputAndOutput(this, getConf(), args);
		if(conf==null) {
			return -1;
		}
		
		conf.setMapperClass(StationMapper.class);
		conf.setMapOutputKeyClass(Text.class);
		conf.setReducerClass(StationReducer.class);
		conf.setOutputKeyClass(NullWritable.class);
		conf.setOutputFormat(StationNameMultipleTextOutputFormat.class);
		
		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception {
		int exitCode = ToolRunner.run(new PartitionByStationUsingMultipleOutputFormat(), args);
		System.exit(exitCode);
	}
}
